package com.bigdata;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class HBaseExecDemo {

    private Configuration conf;
    // HBase 的连接对象
    private Connection conn;
    // HBase 操作DDL的对象
    private Admin admin;


    @Test
    public void createTable() throws Exception {
        /*
        需要确定：
        1. namespace
        2. tableName
        3. columnFamily 1或多个
         */
        String nameSapce = "api_test";
        String table = "student";
        String[] cfs = {"info"};

        TableName tableName = TableName.valueOf(nameSapce, table);
        boolean flag = admin.tableExists(tableName);
        // 判断table是否存在
        if (flag) {
            System.out.println("table 已经存在");
            return;
        }

        // 构建 Table 描述
        TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(tableName);

        // 给table添加列族
        for (String cf : cfs) {
            // 构建cf描述
            ColumnFamilyDescriptorBuilder columnBuilder =
                    ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
            // 列族指定布隆过滤器
            columnBuilder.setBloomFilterType(BloomType.ROW);

            // 列族添加进table
            tableBuilder.setColumnFamily(columnBuilder.build());
        }

        // 创建表格
        admin.createTable(tableBuilder.build());
    }

    @Before
    public void init() throws Exception {
        conf = HBaseConfiguration.create();
        // 设置zk的连接信息
        conf.set("hbase.zookeeper.quorum", "node02,node03,node04");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
    }

    /**
     * 需求2：过滤出学号是以15001001开头的文科学生
     */
    @Test
    public void scanTest2() throws Exception {
        // scan 'ns1:t1', {COLUMNS => ['c1', 'c2'], LIMIT => 10, STARTROW => 'xyz'}
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        // 想要用代码分页查询所有的数据
        Scan scan = new Scan();

        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("15001001"));
        scan.setFilter(prefixFilter);

        /**
         * 创建第二个过滤器，过滤出文科的学生
         */
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("info".getBytes(), "clazz".getBytes(),
                CompareOperator.EQUAL,
                new RegexStringComparator("^文科.*"));

        // 过滤器集合，可以添加到scan对象中 FilterList.Operator.MUST_PASS_ALL
        // 默认就是 and 语义
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);

        filterList.addFilter(prefixFilter);
        filterList.addFilter(singleColumnValueFilter);

        // 追加Filter 过滤器
        scan.setFilter(filterList);

        // 发送scan命令，获取扫描结果
        ResultScanner scanner = table.getScanner(scan);

        processScanner(scanner);

        // 关闭资源
        table.close();
    }

    /**
     * 需求1：过滤出姓于，年纪大于23岁，性别为女，且是理科的学生。
     */
    @Test
    public void scanTest() throws Exception {
        // scan 'ns1:t1', {COLUMNS => ['c1', 'c2'], LIMIT => 10, STARTROW => 'xyz'}
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        // 想要用代码分页查询所有的数据
        Scan scan = new Scan();

        // 过滤出姓于，年纪大于23岁，性别为女，且是理科的学生。
        SingleColumnValueFilter filter1 = new SingleColumnValueFilter(
                "info".getBytes(),
                "name".getBytes(),
                CompareOperator.EQUAL,
                new BinaryPrefixComparator("于".getBytes()));

        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(
                "info".getBytes(),
                "age".getBytes(),
                CompareOperator.GREATER_OR_EQUAL,
                new BinaryComparator("23".getBytes()));

        SingleColumnValueFilter filter3 = new SingleColumnValueFilter(
                "info".getBytes(),
                "gender".getBytes(),
                CompareOperator.EQUAL,
                new BinaryComparator("女".getBytes()));

        SingleColumnValueFilter filter4 = new SingleColumnValueFilter(
                "info".getBytes(),
                "clazz".getBytes(),
                CompareOperator.EQUAL,
                new SubstringComparator("理科"));

        // 过滤器集合，可以添加到scan对象中
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, filter1, filter2,filter3,filter4);

//        filterList.addFilter(filter1);
//        filterList.addFilter(filter2);
//        filterList.addFilter(filter3);
//        filterList.addFilter(filter4);

        // 追加Filter 过滤器
        scan.setFilter(filterList);

        // 发送scan命令，获取扫描结果
        ResultScanner scanner = table.getScanner(scan);

        processScanner(scanner);

        // 关闭资源
        table.close();
    }


    /**
     * 打印扫描结果
     *
     * @param scanner
     */
    private static void processScanner(ResultScanner scanner) {
        try {
            Result result;
            int count = 0;
            while ((result = scanner.next()) != null) {
                processResult(result);
                count++;
            }
            System.out.println("此次扫描到了 " + count + " 行数据");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 打印一个result对象中的数据
     *
     * @param result
     */
    private static void processResult(Result result) {
        // 获取rowkey
        String rowkey = Bytes.toString(result.getRow());

//        String name = new String(result.getValue("info".getBytes(), "name".getBytes()));
        // 列不存在，不会出现空指针
        String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
        String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
        String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
        String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));

        System.out.println("id:" + rowkey + "---name:" + name + "---age:" + age
                + "---gender:" + gender + "---clazz:" + clazz);
    }


    @After
    public void close() throws IOException {
        admin.close();
        conn.close();
    }
}
